package com.myself.storm_test;

import java.util.HashMap;
import java.util.Map;

import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Tuple;
/**
 * 将统计的值汇总输出
 */
@SuppressWarnings("serial")
public class WordCountBolt2 extends BaseBasicBolt {

	Map<String,Integer> counts = new HashMap<String, Integer>();

	public void execute(Tuple tuple, BasicOutputCollector collector) {
		if (tuple.getSourceStreamId() == "SUCCESS_STREAM") {
			System.out.println("prepare to print the statistics");
			for (String key : counts.keySet()) {
				System.out.println(key + "\t" + counts.get(key));
			}
			System.out.println("finish printing");
		} else {

			String word = tuple.getString(0);
			Integer num = tuple.getInteger(1);
			counts.put(word, num);
		}
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
	}
}
